// Use of this source code is governed by a BSD-style license
// that can be found in the License file.
//
// Author: Shuo Chen (chenshuo at chenshuo dot com)

#include "thread_pool.h"
#include "base_exception.h"

#include <assert.h>
#include <stdio.h>

NAMESPACE_START

ThreadPool::ThreadPool(const string &nameArg)
    : mutex_(),
      notEmpty_(mutex_),
      notFull_(mutex_),
      name_(nameArg),
      maxQueueSize_(0),
      running_(false)
{
}

ThreadPool::~ThreadPool()
{
    if (running_)
    {
        stop();
    }
}

void ThreadPool::start(int numThreads)
{
    assert(threads_.empty());
    running_ = true;
    // 存储空间预留
    threads_.reserve(numThreads);
    for (int i = 0; i < numThreads; ++i)
    {
        char id[32];
        snprintf(id, sizeof id, "%d", i + 1);
        threads_.emplace_back(new Thread(
            std::bind(&ThreadPool::runInThread, this), name_ + id));
        threads_[i]->start();
    }
    if (numThreads == 0 && threadInitCallback_)
    {
        threadInitCallback_();
    }
}

void ThreadPool::stop()
{
    {
        // 注意这里的代码块锁
        MutexLockGuard lock(mutex_);
        running_ = false;
        notEmpty_.notifyAll();
    }
    for (auto &thr : threads_)
    {
        thr->join();
    }
}

size_t ThreadPool::queueSize() const
{
    MutexLockGuard lock(mutex_);
    return queue_.size();
}

void ThreadPool::run(Task task)
{
    if (threads_.empty())
    {
        task();
    }
    else
    {
        MutexLockGuard lock(mutex_);
        /* 所有线程都在运行 */
        while (isFull())
        {
            /* 等待空闲的线程 */
            notFull_.wait();
        }
        /* 确认未满 */
        assert(!isFull());
        /* 将task添加到队列中 */
        queue_.push_back(std::move(task));
        /* 提示存在新任务 */
        notEmpty_.notify();
    }
}
/* 获取一个任务 */
ThreadPool::Task ThreadPool::take()
{
    MutexLockGuard lock(mutex_);
    // always use a while-loop, due to spurious wakeup
    // 阻塞等待任务队列
    while (queue_.empty() && running_)
    {
        notEmpty_.wait();
    }
    // 任务函数
    Task task;
    // 
    if (!queue_.empty())
    {
        // 获取第一个任务
        task = queue_.front();
        queue_.pop_front();
        // 发送信号显示未满
        if (maxQueueSize_ > 0)
        {
            notFull_.notify();
        }
    }
    return task;
}

bool ThreadPool::isFull() const
{
    mutex_.assertLocked();
    return maxQueueSize_ > 0 && queue_.size() >= maxQueueSize_;
}
/* 开启线程 */
void ThreadPool::runInThread()
{
    try
    {
        // 执行初始化回调任务
        if (threadInitCallback_)
        {
            threadInitCallback_();
        }
        /* 不断的进行任务的获取 */
        while (running_)
        {
            Task task(take());
            // 执行任务
            if (task)
            {
                task();
            }
        }
    }
    catch (const Exception &ex)
    {
        fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());
        fprintf(stderr, "reason: %s\n", ex.what());
        fprintf(stderr, "stack trace: %s\n", ex.stackTrace());
        abort();
    }
    catch (const std::exception &ex)
    {
        fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());
        fprintf(stderr, "reason: %s\n", ex.what());
        abort();
    }
    catch (...)
    {
        fprintf(stderr, "unknown exception caught in ThreadPool %s\n", name_.c_str());
        throw; // rethrow
    }
}
NAMESPACE_END
